mini_blockchain/
├── models/
│   ├── __init__.py
│   ├── transaction.py
│   └── block.py
├── utils/
│   ├── __init__.py
│   └── hash.py
├── main.py
└── tests/

from dataclasses import dataclass, asdict

@dataclass
class Transaction:
    from_addr: str
    to_addr: str
    amount: float
    signature: str = ""  # 简化版先留占位符

    def to_dict(self) -> dict:
        return asdict(self)

from dataclasses import dataclass, field
from typing import List
import time

@dataclass
class Block:
    index: int
    timestamp: float
    transactions: List[Transaction]
    prevHash: str
    nonce: int = 0
    hash: str = field(default="", compare=False)

"index|timestamp|transactions_str|prevHash|nonce"

import hashlib
import json

def calculate_hash(block: Block) -> str:
    # 将交易列表转为标准化 JSON 字符串，确保顺序一致
    tx_str = json.dumps([tx.to_dict() for tx in block.transactions], sort_keys=True, separators=(',', ':'))
    # 按固定顺序拼接区块头信息
    raw = f"{block.index}|{block.timestamp}|{tx_str}|{block.prevHash}|{block.nonce}"
    # 双重 SHA-256
    first = hashlib.sha256(raw.encode('utf-8')).digest()
    second = hashlib.sha256(first).hexdigest()
    return second

class Blockchain:
    def __init__(self, difficulty: int = 2):
        self.chain: List[Block] = []
        self.difficulty = difficulty
        # 创建并追加创世块
        genesis = self._create_genesis_block()
        self.chain.append(genesis)

    @property
    def latest_block(self) -> Block:
        return self.chain[-1]

class Blockchain:
    # ... 接上文 ...

    def is_chain_valid(self) -> bool:
        for i in range(1, len(self.chain)):
            curr = self.chain[i]
            prev = self.chain[i - 1]

            # 规则1：连续性验证
            if curr.index != prev.index + 1:
                print(f"[验证失败] 区块 {curr.index} 索引不连续")
                return False

            # 规则2：前向哈希匹配
            if curr.prevHash != prev.hash:
                print(f"[验证失败] 区块 {curr.index} prevHash 与前一区块不匹配")
                return False

            # 规则3：哈希有效性
            if calculate_hash(curr) != curr.hash:
                print(f"[验证失败] 区块 {curr.index} 哈希被篡改")
                return False

        return True

    def add_block(self, block: Block) -> bool:
        """在通过链接验证后将新区块加入链"""
        # 简单校验：索引和 prevHash 匹配当前链尾
        if block.index != self.latest_block.index + 1:
            print("[拒绝] 新块索引不连续")
            return False
        if block.prevHash != self.latest_block.hash:
            print("[拒绝] 新块 prevHash 不匹配链尾")
            return False
        self.chain.append(block)
        return True

class Blockchain:
    # ... 接上文 ...

    def _create_genesis_block(self) -> Block:
        genesis = Block(
            index=0,
            timestamp=time.time(),
            transactions=[],
            prevHash="0" * 64,  # 64 个零，象征"无前一区块"
            nonce=0,
        )
        # 为简化，创世块直接计算一次哈希，不走挖矿流程
        genesis.hash = calculate_hash(genesis)
        return genesis

def mine_block(block: Block, difficulty: int) -> Block:
    """
    PoW 挖矿：暴力调整 nonce，直到 hash 的前 difficulty 个字符全为 '0'。
    """
    target = "0" * difficulty
    block.nonce = 0
    while True:
        block.hash = calculate_hash(block)
        if block.hash[:difficulty] == target:
            break
        block.nonce += 1
        # 教学提示：nonce 可能非常大；在真实网络中还会配合
        # 修改 coinbase 交易的 extraNonce、修改时间戳等策略
    return block

class Blockchain:
    TARGET_BLOCK_TIME = 5.0  # 目标出块时间 5 秒

    def __init__(self, difficulty: int = 2):
        self.chain: List[Block] = []
        self.difficulty = max(1, difficulty)
        self.chain.append(self._create_genesis_block())

    def adjust_difficulty(self) -> None:
        """
        每产 5 个区块回顾一次，根据实际平均出块时间调整难度。
        """
        if len(self.chain) < 6:
            return  # 创世块 + 不足5个新区块，暂不调整
        if (len(self.chain) - 1) % 5 != 0:
            return  # 不是回顾窗口的边界

        # 计算最近 5 个区块（不包括创世块）的平均出块时间
        recent = self.chain[-5:]
        intervals = [recent[i].timestamp - self.chain[self.chain.index(recent[i]) - 1].timestamp
                     for i in range(len(recent))]
        avg_actual = sum(intervals) / len(intervals)

        ratio = avg_actual / self.TARGET_BLOCK_TIME
        new_diff = int(self.difficulty * ratio)
        new_diff = max(1, new_diff)

        print(f"[难度调整] 最近5块平均间隔 {avg_actual:.3f}s, 目标 {self.TARGET_BLOCK_TIME}s, "
              f"旧难度 {self.difficulty} -> 新难度 {new_diff}")
        self.difficulty = new_diff

import hashlib
import json
import time
from dataclasses import dataclass, field, asdict
from typing import List

# ============= 模型定义 =============

@dataclass
class Transaction:
    from_addr: str
    to_addr: str
    amount: float
    signature: str = ""

    def to_dict(self) -> dict:
        return asdict(self)

@dataclass
class Block:
    index: int
    timestamp: float
    transactions: List[Transaction]
    prevHash: str
    nonce: int = 0
    hash: str = field(default="", compare=False)

# ============= 哈希工具 =============

def calculate_hash(block: Block) -> str:
    tx_str = json.dumps([tx.to_dict() for tx in block.transactions], sort_keys=True, separators=(',', ':'))
    raw = f"{block.index}|{block.timestamp}|{tx_str}|{block.prevHash}|{block.nonce}"
    first = hashlib.sha256(raw.encode('utf-8')).digest()
    return hashlib.sha256(first).hexdigest()

# ============= 挖矿逻辑 =============

def mine_block(block: Block, difficulty: int) -> Block:
    target = "0" * difficulty
    block.nonce = 0
    while True:
        block.hash = calculate_hash(block)
        if block.hash[:difficulty] == target:
            break
        block.nonce += 1
    return block

# ============= 区块链 =============

class Blockchain:
    TARGET_BLOCK_TIME = 5.0

    def __init__(self, difficulty: int = 2):
        self.chain: List[Block] = []
        self.difficulty = max(1, difficulty)
        self.chain.append(self._create_genesis_block())

    @property
    def latest_block(self) -> Block:
        return self.chain[-1]

    def _create_genesis_block(self) -> Block:
        b = Block(index=0, timestamp=time.time(), transactions=[], prevHash="0" * 64, nonce=0)
        b.hash = calculate_hash(b)
        return b

    def is_chain_valid(self) -> bool:
        for i in range(1, len(self.chain)):
            curr, prev = self.chain[i], self.chain[i - 1]
            if curr.index != prev.index + 1:
                return False
            if curr.prevHash != prev.hash:
                return False
            if calculate_hash(curr) != curr.hash:
                return False
        return True

    def add_block(self, block: Block) -> bool:
        if block.index != self.latest_block.index + 1:
            return False
        if block.prevHash != self.latest_block.hash:
            return False
        self.chain.append(block)
        return True

    def adjust_difficulty(self) -> None:
        if len(self.chain) < 6:
            return
        if (len(self.chain) - 1) % 5 != 0:
            return
        recent = self.chain[-5:]
        intervals = []
        for i in range(len(recent)):
            idx = self.chain.index(recent[i])
            intervals.append(recent[i].timestamp - self.chain[idx - 1].timestamp)
        avg_actual = sum(intervals) / len(intervals)
        new_diff = max(1, int(self.difficulty * (avg_actual / self.TARGET_BLOCK_TIME)))
        print(f"\n>>> [难度调整] 5块平均间隔 {avg_actual:.3f}s | 旧难度 {self.difficulty} -> 新难度 {new_diff}\n")
        self.difficulty = new_diff

# ============= 主演示 =============

def main():
    print("=== 迷你区块链 PoW 挖矿与动态难度调整演示 ===\n")
    bc = Blockchain(difficulty=2)
    print(f"[创世块] index=0, hash={bc.chain[0].hash[:16]}...")

    for idx in range(1, 13):
        tx = Transaction(from_addr="Alice", to_addr="Bob", amount=1.0 * idx)
        new_block = Block(
            index=idx,
            timestamp=0,  # 先占位，挖矿前不设定
            transactions=[tx],
            prevHash=bc.latest_block.hash,
        )

        new_block.timestamp = time.time()
        start = time.time()
        mine_block(new_block, bc.difficulty)
        elapsed = time.time() - start
        bc.add_block(new_block)

        print(f"[出块] index={idx:3d} | 耗时 {elapsed:.4f}s | "
              f"nonce={new_block.nonce:>8d} | hash={new_block.hash[:16]}... | 难度={bc.difficulty}")

        bc.adjust_difficulty()

    print(f"\n=== 全链验证结果: {bc.is_chain_valid()} ===")
    print(f"=== 总区块数: {len(bc.chain)} ===")

if __name__ == "__main__":
    main()

state = {
    "Alice": 100.0,
    "Bob":    0.0
}
nonces = {
    "Alice": 0,
    "Bob":   0
}

import hashlib, json
from dataclasses import dataclass, asdict, field
from typing import Optional

@dataclass
class Transaction:
    sender: str
    recipient: str
    amount: float
    nonce: int = 0
    signature: str = ""

    def to_dict(self) -> dict:
        # 用有序 dict 保证序列化一致性，避免哈希波动
        return {
            "sender": self.sender,
            "recipient": self.recipient,
            "amount": self.amount,
            "nonce": self.nonce,
        }

    def hash(self) -> str:
        raw = json.dumps(self.to_dict(), sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(raw.encode()).hexdigest()

from ecdsa import SigningKey, VerifyingKey, SECP256k1
from ecdsa.util import sigencode_der, sigdecode_der

def generate_keypair():
    sk = SigningKey.generate(curve=SECP256k1)
    vk = sk.verifying_key
    return sk, vk

def sign_tx(sk: SigningKey, tx_hash: str) -> str:
    return sk.sign(tx_hash.encode(), sigencode=sigencode_der).hex()

def verify_tx(vk: VerifyingKey, tx_hash: str, signature_hex: str) -> bool:
    try:
        return vk.verify(
            bytes.fromhex(signature_hex),
            tx_hash.encode(),
            sigdecode=sigdecode_der,
        )
    except Exception:
        return False

def validate_transaction(tx: Transaction, state: dict, nonces: dict) -> bool:
    # 1. 余额足够
    if state.get(tx.sender, 0) < tx.amount:
        return False

    # 2. nonce 严格递增（不允许跳号）
    if nonces.get(tx.sender, 0) != tx.nonce:
        return False

    # 3. 签名验证（要求 sender 是十六进制公钥地址）
    try:
        vk = VerifyingKey.from_string(bytes.fromhex(tx.sender), curve=SECP256k1)
        tx_hash = tx.hash()
        if not verify_tx(vk, tx_hash, tx.signature):
            return False
    except Exception:
        return False

    return True

def apply_transaction(tx: Transaction, state: dict, nonces: dict):
    if not validate_transaction(tx, state, nonces):
        return False
    state[tx.sender] = state.get(tx.sender, 0) - tx.amount
    state[tx.recipient] = state.get(tx.recipient, 0) + tx.amount
    nonces[tx.sender] = tx.nonce + 1
    return True

def create_coinbase_tx(miner_address: str, reward: float = 50.0) -> Transaction:
    return Transaction(
        sender="COINBASE",
        recipient=miner_address,
        amount=reward,
        nonce=0,
        signature="",
    )

peers = set()          # 已知邻居的 URL，如 {"http://localhost:5001", ...}
blockchain = []        # 本地区块链副本
state = {}             # 全局余额状态
nonces = {}            # 地址 nonce 计数器
mempool = []           # 待打包交易列表
node_id = str(uuid.uuid4())[:8]

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/nodes/register", methods=["POST"])
def register_node():
    data = request.get_json()
    node_url = data.get("url")
    if node_url and node_url not in peers:
        peers.add(node_url)
    return jsonify({"message": "registered", "peers": list(peers)})

@app.route("/nodes", methods=["GET"])
def get_peers():
    return jsonify({"peers": list(peers)})

import requests
from concurrent.futures import ThreadPoolExecutor

def broadcast(msg_type: str, data: dict):
    payload = {"type": msg_type, "data": data, "timestamp": time.time()}
    with ThreadPoolExecutor(max_workers=10) as pool:
        for peer in peers:
            pool.submit(requests.post, f"{peer}/p2p", json=payload, timeout=2)

@app.route("/p2p", methods=["POST"])
def handle_p2p_message():
    msg = request.get_json()
    msg_type = msg["type"]
    data = msg["data"]

    if msg_type == "NEW_TRANSACTION":
        tx = Transaction(**data)
        if validate_transaction(tx, state, nonces):
            mempool.append(tx)
            broadcast("NEW_TRANSACTION", data)   # 继续转发

    elif msg_type == "NEW_BLOCK":
        handle_incoming_block(data)

    elif msg_type == "REQUEST_CHAIN":
        return jsonify({"chain": [b.__dict__ for b in blockchain], "length": len(blockchain)})

    elif msg_type == "RESPONSE_CHAIN":
        remote_chain = data["chain"]
        resolve_conflicts(remote_chain)

    return jsonify({"status": "ok"})

def resolve_conflicts(remote_chain: list) -> bool:
    # 只接受严格更长的有效链
    if len(remote_chain) <= len(blockchain):
        return False

    # 验证远程链的 PoW 与前序哈希
    for i in range(1, len(remote_chain)):
        prev = remote_chain[i - 1]
        cur = remote_chain[i]
        cur_hash = calculate_block_hash(cur)
        if cur["prev_hash"] != calculate_block_hash(prev):
            return False
        if not cur_hash.startswith("0" * difficulty):
            return False

    # 替换本地链，重新计算状态
    blockchain.clear()
    blockchain.extend(remote_chain)
    rebuild_state()
    return True

def rebuild_state():
    """从创世块开始逐块重放交易，重建 state 和 nonces"""
    state.clear()
    nonces.clear()
    for block in blockchain:
        for tx_data in block["transactions"]:
            tx = Transaction(**tx_data)
            apply_transaction(tx, state, nonces)

@app.route("/blocks", methods=["GET"])
def get_blocks():
    return jsonify([b.__dict__ for b in blockchain])

@app.route("/blocks/<int:index>", methods=["GET"])
def get_block(index):
    if 0 <= index < len(blockchain):
        return jsonify(blockchain[index].__dict__)
    return jsonify({"error": "block not found"}), 404

@app.route("/transactions/new", methods=["POST"])
def new_transaction():
    data = request.get_json()
    tx = Transaction(
        sender=data["sender"],
        recipient=data["recipient"],
        amount=data["amount"],
        nonce=data.get("nonce", 0),
        signature=data.get("signature", ""),
    )
    if validate_transaction(tx, state, nonces):
        mempool.append(tx)
        broadcast("NEW_TRANSACTION", tx.to_dict())
        return jsonify({"message": "transaction accepted", "index": len(blockchain)})
    return jsonify({"error": "invalid transaction"}), 400

@app.route("/mine", methods=["GET", "POST"])
def mine():
    if not mempool:
        # 没有用户交易时，至少包含 coinbase 奖励
        reward_tx = create_coinbase_tx(node_id)
        mempool.append(reward_tx)

    new_block = mine_block(list(mempool), blockchain, difficulty)
    if new_block:
        mempool.clear()
        broadcast("NEW_BLOCK", new_block.__dict__)
        return jsonify(new_block.__dict__)
    return jsonify({"error": "mining failed"}), 500

@app.route("/balance/<address>", methods=["GET"])
def get_balance(address):
    return jsonify({"address": address, "balance": state.get(address, 0)})

<!DOCTYPE html>
<html>
<head><title>Mini Blockchain Explorer</title></head>
<body>
  <h1>🔗 Mini Blockchain Explorer</h1>
  <div id="stats">
    <span id="height">Height: --</span>
    <span id="txcount">Txs: --</span>
    <span id="peers">Peers: --</span>
  </div>

  <h2>发起交易</h2>
  <form id="txForm">
    <input name="sender" placeholder="发送方公钥（hex）" size="50" />
    <input name="recipient" placeholder="接收方公钥（hex）" size="50" />
    <input name="amount" type="number" step="0.01" placeholder="金额" />
    <button type="submit">发送交易</button>
  </form>

  <h2>区块链</h2>
  <table border="1" id="chainTable">
    <thead>
      <tr>
        <th>#</th><th>Time</th><th>Miner</th><th>Txs</th><th>Hash (前 16 位)</th>
      </tr>
    </thead>
    <tbody id="chainBody"></tbody>
  </table>

  <script>
    const API = "/blocks";

    function renderBlocks(blocks) {
      const body = document.getElementById("chainBody");
      body.innerHTML = blocks.map(b => `
        <tr>
          <td>${b.index}</td>
          <td>${new Date(b.timestamp * 1000).toLocaleTimeString()}</td>
          <td>${(b.miner || "?").slice(0, 10)}...</td>
          <td>${b.transactions?.length || 0}</td>
          <td>${b.hash.slice(0, 16)}</td>
        </tr>
      `).join("");
      document.getElementById("height").textContent = `Height: ${blocks.length - 1}`;
    }

    function fetchChain() {
      fetch(API).then(r => r.json()).then(renderBlocks).catch(console.error);
    }

    document.getElementById("txForm").addEventListener("submit", e => {
      e.preventDefault();
      const fd = new FormData(e.target);
      fetch("/transactions/new", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify(Object.fromEntries(fd)),
      }).then(r => r.json()).then(console.log);
    });

    setInterval(fetchChain, 3000);
    fetchChain();
  </script>
</body>
</html>

# 终端 1：Bootstrap 节点
python node.py --port 5000 --bootstrap http://localhost:5000

# 终端 2：节点 B
python node.py --port 5001 --bootstrap http://localhost:5000

# 终端 3：节点 C
python node.py --port 5002 --bootstrap http://localhost:5000

# network_sim.py —— 多进程一键启动示例
import subprocess, time

nodes = [
    ("Alice", 5000),
    ("Bob",   5001),
    ("Carol", 5002),
]
procs = []
for name, port in nodes:
    p = subprocess.Popen(["python", "node.py",
        "--port", str(port),
        "--bootstrap", "http://localhost:5000",
        "--name", name,
    ])
    procs.append(p)
    time.sleep(0.5)

print("网络启动完毕。按 Enter 关闭所有节点。")
input()
for p in procs:
    p.kill()

# 1. 查询余额
curl http://localhost:5000/balance/Alice

# 2. 生成密钥对并签名（Python 伪代码，实际用编程方式调用）
# sk, vk = generate_keypair()
# tx = Transaction(sender=vk.hex(), recipient="Bob", amount=5, nonce=0)
# sig = sign_tx(sk, tx.hash())
# tx.signature = sig

# 3. 发送签名交易
curl -X POST http://localhost:5000/transactions/new \
  -H "Content-Type: application/json" \
  -d '{"sender":"<公钥hex>","recipient":"Bob","amount":5,"nonce":0,"signature":"<签名hex>"}'

# 4. 触发挖矿
curl http://localhost:5000/mine

# 5. 检查区块和余额
curl http://localhost:5000/blocks/1
curl http://localhost:5000/balance/Bob

#!/bin/bash
# scripts/run_3_nodes.sh
python node.py --port 5001 --data-dir ./data/node1 &
python node.py --port 5002 --data-dir ./data/node2 --bootstrap http://127.0.0.1:5001 &
python node.py --port 5003 --data-dir ./data/node3 --bootstrap http://127.0.0.1:5001 &
wait

# tests/test_integration.py
import pytest
from blockchain import Blockchain, Transaction
from crypto import sign_tx, verify_tx

def test_chain_validity(chain_with_blocks):
    """验证链上每个区块的哈希与前块Hash字段连续性"""
    for i in range(1, len(chain_with_blocks.chain)):
        block = chain_with_blocks.chain[i]
        prev_block = chain_with_blocks.chain[i - 1]
        assert block.prev_hash == prev_block.hash
        assert block.hash == block.calculate_hash()

def test_tx_signature_verification(key_pair):
    """构造合法/篡改交易，断言验证通过/拒绝"""
    priv_key, pub_key = key_pair
    tx = Transaction(from_addr=pub_key, to="Bob", amount=5, nonce=1)
    tx.signature = sign_tx(tx.hash(), priv_key)
    assert verify_tx(tx, pub_key) == True

    # 篡改交易内容
    tx.amount = 100
    assert verify_tx(tx, pub_key) == False

def test_difficulty_adjustment(mock_time):
    """构造跨难度周期的时间戳，验证目标值重新计算"""
    bc = Blockchain(difficulty=4)
    # 模拟出块时间远快于预期（10秒出了20个区块）
    for i in range(20):
        bc.mine_block([], timestamp=bc.last_block.timestamp + 0.5)
    # 验证难度已增加（目标值变小）
    assert bc.target < (1 << (256 - 4))

# tests/conftest.py
import pytest
from ecdsa import SigningKey, SECP256k1

@pytest.fixture
def key_pair():
    sk = SigningKey.generate(curve=SECP256k1)
    vk = sk.verifying_key
    return sk, vk

@pytest.fixture
def mock_time():
    import time
    original = time.time
    time.time = lambda: 1234567890
    yield
    time.time = original

class UTXOSet:
    def __init__(self):
        self.utxos: dict[OutPoint, TxOutput] = {}

    def apply_tx(self, tx: Transaction) -> bool:
        """验证并应用一笔交易：删除已花费输出，添加新输出"""
        if not self.validate_inputs(tx.inputs):
            return False
        for inp in tx.inputs:
            outpoint = OutPoint(inp.prev_txid, inp.output_index)
            del self.utxos[outpoint]           # 删除已花费输出
        for idx, outp in enumerate(tx.outputs):
            outpoint = OutPoint(tx.txid, idx)
            self.utxos[outpoint] = outp        # 添加新输出
        return True

    def revert_tx(self, tx: Transaction):
        """链重组时回滚：恢复旧输出，删除新输出"""
        for idx, outp in enumerate(tx.outputs):
            outpoint = OutPoint(tx.txid, idx)
            del self.utxos[outpoint]
        for inp in tx.inputs:
            outpoint = OutPoint(inp.prev_txid, inp.output_index)
            self.utxos[outpoint] = ...         # 从存档中恢复